package com.gandong8.tablestore.tool.util;

import com.alicloud.openservices.tablestore.model.*;
import com.alicloud.openservices.tablestore.model.search.CreateSearchIndexRequest;
import com.alicloud.openservices.tablestore.model.search.FieldSchema;
import com.alicloud.openservices.tablestore.model.search.FieldType;
import com.alicloud.openservices.tablestore.model.search.IndexSchema;
import com.fasterxml.jackson.databind.PropertyNamingStrategy;
import com.gandong8.tablestore.tool.annotation.TsField;
import com.gandong8.tablestore.tool.annotation.TsTable;
import com.gandong8.tablestore.tool.domain.TsFieldTypeEnum;
import com.gandong8.tablestore.tool.exception.TypeNotSupportException;

import java.beans.PropertyDescriptor;
import java.lang.reflect.Field;
import java.lang.reflect.Method;
import java.math.BigDecimal;
import java.util.*;

public  class TableStoreUtil {
    private static Map<Class<?> , Map<String, Field>> map = new HashMap<>();

    public static CreateTableRequest buildCreateTable(Class<?> clazz) {
        TsTable table = clazz.getAnnotation(TsTable.class);
        TableMeta tableMeta = new TableMeta(table.name());
        Field[] fields = clazz.getDeclaredFields();
        String indexName = table.secondIndexName();
        ArrayList<IndexMeta> indexMetas = new ArrayList<>();
        IndexMeta indexMeta = null;
        if(!"".equals(indexName)) {
            indexMeta = new IndexMeta(indexName);
        }
        for (Field field : fields) {
            TsField tsField = field.getAnnotation(TsField.class);
            if (tsField == null) {
                continue;
            }
            String filedName = getTsFieldName(table.propertySnakeCase(), field.getName(), tsField);
            if(tsField.primaryKey()) {
                String typeName = field.getType().getTypeName();
                TsFieldTypeEnum typeEnum = TsFieldTypeEnum.get(typeName);
                if(TsFieldTypeEnum.STRING.equals(typeEnum)) {
                    tableMeta.addPrimaryKeyColumn(new PrimaryKeySchema(filedName, PrimaryKeyType.STRING));
                }
                if(TsFieldTypeEnum.INTEGER.equals(typeEnum)) {
                    tableMeta.addPrimaryKeyColumn(new PrimaryKeySchema(filedName, PrimaryKeyType.INTEGER));
                }
            }
            if(indexMeta != null && tsField.secondIndex()) {
                if(tsField.primaryKey()) {
                    indexMeta.addPrimaryKeyColumn(filedName);
                } else {
                    tableMeta.addDefinedColumn(new DefinedColumnSchema(filedName, Objects.requireNonNull(getDefinedColumnType(field))));
                    indexMeta.addDefinedColumn(filedName);
                }
            }
        }
        if(indexMeta != null) {
            indexMetas.add(indexMeta);
        }
        TableOptions tableOptions = new TableOptions(table.timeToLive(), table.maxVersions());
        return new CreateTableRequest(tableMeta, tableOptions, indexMetas);
    }

    public static CreateSearchIndexRequest buildSearchIndex(Class<?> clazz) {
        CreateSearchIndexRequest request = new CreateSearchIndexRequest();
        TsTable table = clazz.getAnnotation(TsTable.class);
        request.setTableName(table.name());
        if(table.multiIndexName().equals("")) {
            return null;
        }
        request.setIndexName(table.multiIndexName());
        IndexSchema indexSchema = new IndexSchema();
        Field[] fields = clazz.getDeclaredFields();
        List<FieldSchema> fieldSchemaList = new ArrayList<>();
        for (Field field : fields) {
            TsField tsField = field.getAnnotation(TsField.class);
            if (tsField == null) {
                continue;
            }
            if(!tsField.multiIndex()) {
                continue;
            }
            String filedName = getTsFieldName(table.propertySnakeCase(), field.getName(), tsField);
            FieldType fieldType = getFieldType(field);
            if(fieldType == null) {
                continue;
            }
            fieldSchemaList.add( new FieldSchema(filedName, fieldType)
                    .setIndex(true) //设置开启索引。
                    .setEnableSortAndAgg(true));//设置开启排序与统计聚合功能。
        }
        if(fieldSchemaList.size() < 1) {
            return null;
        }
        indexSchema.setFieldSchemas(fieldSchemaList);
        request.setIndexSchema(indexSchema);
        return request;
    }

    private static FieldType getFieldType(Field field) {
        String typeName = field.getType().getTypeName();
        TsFieldTypeEnum typeEnum = TsFieldTypeEnum.get(typeName);
        if(TsFieldTypeEnum.STRING.equals(typeEnum)) {
            return FieldType.TEXT;
        }
        if(TsFieldTypeEnum.INTEGER.equals(typeEnum)) {
            return FieldType.LONG;
        }
        if(TsFieldTypeEnum.DATE.equals(typeEnum)) {
            return FieldType.LONG;
        }
        if(TsFieldTypeEnum.BOOLEAN.equals(typeEnum)) {
            return FieldType.BOOLEAN;
        }
        if(TsFieldTypeEnum.DOUBLE.equals(typeEnum)) {
            return FieldType.DOUBLE;
        }
        return null;
    }

    private static DefinedColumnType getDefinedColumnType(Field field) {
        String typeName = field.getType().getTypeName();
        TsFieldTypeEnum typeEnum = TsFieldTypeEnum.get(typeName);
        if(TsFieldTypeEnum.STRING.equals(typeEnum)) {
            return DefinedColumnType.STRING;
        }
        if(TsFieldTypeEnum.INTEGER.equals(typeEnum)) {
            return DefinedColumnType.INTEGER;
        }
        if(TsFieldTypeEnum.DATE.equals(typeEnum)) {
            return DefinedColumnType.INTEGER;
        }
        if(TsFieldTypeEnum.BOOLEAN.equals(typeEnum)) {
            return DefinedColumnType.BOOLEAN;
        }
        if(TsFieldTypeEnum.DOUBLE.equals(typeEnum)) {
            return DefinedColumnType.DOUBLE;
        }
        if(TsFieldTypeEnum.BYTE.equals(typeEnum)) {
            return DefinedColumnType.BINARY;
        }
        return null;
    }

    public static String getTableName(Class<?> clazz) {
        //todo cache
        TsTable table = clazz.getAnnotation(TsTable.class);
        return table.name();
    }

    private static String getTsFieldName(Class<?> clazz, String filedName, TsField tsField) {
        return getTsFieldName(isSnake(clazz), filedName, tsField);
    }

    private static String getTsFieldName(Boolean isSnake, String filedName, TsField tsField) {
        if(!"".equals(tsField.name())) {
            return tsField.name();
        }
        if(isSnake) {
            return translateFieldName(filedName);
        }
        return filedName;
    }


    public static <T> T parseStreamRecord(Class<?> clazz, StreamRecord streamRecord) {
        T obj = null;
        try {
            obj = (T) clazz.newInstance();
            parseStreamRecord(obj, streamRecord);
        } catch (Exception e) {
            e.printStackTrace();
        }
        return obj;
    }

    public static <T> T parseRow(Class<?> clazz, Row row) {
        T obj = null;
        try {
            obj = (T) clazz.newInstance();
            parseRow(obj, row);
        } catch (Exception e) {
            e.printStackTrace();
        }
        return obj;
    }

    public static void parseStreamRecord(Object obj, StreamRecord streamRecord) {
        PrimaryKeyColumn[] primaryKeyColumns = streamRecord.getPrimaryKey().getPrimaryKeyColumns();
        List<RecordColumn> recordColumns = streamRecord.getColumns();
        List<Column> columnList = new ArrayList<>();
        for(RecordColumn recordColumn:recordColumns) {
            columnList.add(recordColumn.getColumn());
        }
        parseColumn(obj, primaryKeyColumns, columnList);
    }

    public static void parseRow(Object obj, Row row) {
        Column[] columns = row.getColumns();
        PrimaryKeyColumn[] primaryKeyColumns = row.getPrimaryKey().getPrimaryKeyColumns();
        parseColumn(obj, primaryKeyColumns, Arrays.asList(columns));
    }

    public static void  parseColumn(Object obj, PrimaryKeyColumn[] primaryKeyColumns, List<Column> columns) {
        Class clazz = obj.getClass();
        Field[] fields = clazz.getDeclaredFields();
        Map<String, Field> nameMap = new HashMap<>();
        if(map.containsKey(clazz)) {
            nameMap = map.get(clazz);
        } else {
            for (Field field : fields) {
                TsField tsField = field.getAnnotation(TsField.class);
                if (tsField == null) {
                    continue;
                }
                nameMap.put(getTsFieldName(clazz, field.getName(), tsField), field);
            }
            map.put(clazz, nameMap);
        }
        for(PrimaryKeyColumn primaryKeyColumn:primaryKeyColumns) {
            Field field = nameMap.get(primaryKeyColumn.getName());
            try {
                Method method = new PropertyDescriptor(field.getName(), clazz).getWriteMethod();
                method.invoke(obj, convertColumnToObj(field.getType(), primaryKeyColumn.getValue().toColumnValue()));
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
        for(Column column:columns) {
            Field field = nameMap.get(column.getName());
            try {
            Method method = new PropertyDescriptor(field.getName(), clazz).getWriteMethod();
            method.invoke(obj, convertColumnToObj(field.getType(), column.getValue()));
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }

    public static void buildQuery(Class clazz, RowQueryCriteria criteria) {
        Field[] fields = clazz.getDeclaredFields();
        for (Field field : fields) {
            TsField tsField = field.getAnnotation(TsField.class);
            if (tsField == null) {
                continue;
            }
            if (tsField.primaryKey()) {
                continue;
            }
            criteria.addColumnsToGet(getTsFieldName(clazz, field.getName(), tsField));
        }
    }

    public static RowUpdateChange buildUpdate(Object obj) {
        Class clazz = obj.getClass();
        RowUpdateChange rowUpdateChange = new RowUpdateChange(getTableName(clazz), parsePrimaryKey(obj));
        Field[] fields = clazz.getDeclaredFields();
        for (Field field : fields) {
            TsField tsField = field.getAnnotation(TsField.class);
            if (tsField == null || tsField.primaryKey()) {
                continue;
            }
            field.setAccessible(true);
            Object filedObject;
            try {
                filedObject = field.get(obj);
                if(filedObject == null) {
                    continue;
                }
            } catch (IllegalAccessException e) {
                e.printStackTrace();
                continue;
            }
            String filedName = getTsFieldName(clazz, field.getName(), tsField);
            ColumnValue columnValue = convertColumnValue(field.getType(), filedObject);
            if (columnValue != null) {
                rowUpdateChange.put(new Column(filedName, columnValue));
            }
        }
        return rowUpdateChange;
    }

    public static RowPutChange buildPut(Object obj) {
        Class clazz = obj.getClass();
        RowPutChange putChange = new RowPutChange(getTableName(clazz), parsePrimaryKey(obj));
        Field[] fields = clazz.getDeclaredFields();
        for (Field field : fields) {
            TsField tsField = field.getAnnotation(TsField.class);
            if (tsField == null || tsField.primaryKey()) {
                continue;
            }
            field.setAccessible(true);
            Object filedObject;
            try {
                filedObject = field.get(obj);
                if(filedObject == null) {
                    continue;
                }
            } catch (IllegalAccessException e) {
                e.printStackTrace();
                continue;
            }
            String filedName = getTsFieldName(clazz, field.getName(), tsField);
            ColumnValue columnValue = convertColumnValue(field.getType(), filedObject);
            if (columnValue != null) {
                putChange.addColumn(new Column(filedName, columnValue));
            }
        }
        return putChange;
    }

    public static PrimaryKey parsePrimaryKey(Object obj) {
        Class clazz = obj.getClass();
        Field[] fields = clazz.getDeclaredFields();
        PrimaryKeyBuilder primaryKeyBuilder = PrimaryKeyBuilder.createPrimaryKeyBuilder();
        boolean hasPrimary = false;
        for (Field field : fields) {
            TsField tsField = field.getAnnotation(TsField.class);
            if (tsField == null) {
                continue;
            }
            if (!tsField.primaryKey()) {
                continue;
            }
            field.setAccessible(true);
            String filedName = getTsFieldName(clazz, field.getName(), tsField);
            Object filedObject;
            try {
                filedObject = field.get(obj);
            } catch (IllegalAccessException e) {
                e.printStackTrace();
                continue;
            }
            PrimaryKeyValue primaryKeyValue = convertPrimaryKeyValue(field.getType(), filedObject);
            if (primaryKeyValue == null) {
                if (tsField.autoIncrement()) {
                    primaryKeyBuilder.addPrimaryKeyColumn(filedName, PrimaryKeyValue.AUTO_INCREMENT);
                }
                continue;
            }
            primaryKeyBuilder.addPrimaryKeyColumn(filedName,  primaryKeyValue);
            hasPrimary = true;
        }
        return hasPrimary ? primaryKeyBuilder.build() : null;
    }

    private static ColumnValue convertColumnValue(Class<?> type, Object value) {
        if (value == null) {
            return null;
        }
        String typeName = type.getTypeName();
        TsFieldTypeEnum typeEnum = TsFieldTypeEnum.get(typeName);
        if(typeEnum == null) {
            throw new TypeNotSupportException(typeName);
        }
        if(typeEnum.equals(TsFieldTypeEnum.DATE)) {
            Date date = (Date) value;
            return ColumnValue.fromLong(date.getTime());
        }
        if(typeEnum.equals(TsFieldTypeEnum.BOOLEAN)) {
            return ColumnValue.fromBoolean(Boolean.parseBoolean(String.valueOf(value)));
        }
        if(typeEnum.equals(TsFieldTypeEnum.BYTE)) {
            return ColumnValue.fromBinary((byte[]) value);
        }
        if(typeEnum.equals(TsFieldTypeEnum.DOUBLE)) {
            return ColumnValue.fromDouble(Double.parseDouble(String.valueOf(value)));
        }
        if(typeEnum.equals(TsFieldTypeEnum.LONG)) {
            return ColumnValue.fromLong(Long.parseLong(String.valueOf(value)));
        }
        if(typeEnum.equals(TsFieldTypeEnum.BIGDECIMAL)) {
            BigDecimal res = (BigDecimal) value;
            return ColumnValue.fromDouble(res.doubleValue());
        }
        if(typeEnum.equals(TsFieldTypeEnum.INTEGER)) {
            return ColumnValue.fromLong(Long.parseLong(String.valueOf(value)));
        }
        if(typeEnum.equals(TsFieldTypeEnum.STRING)) {
            return ColumnValue.fromString(String.valueOf(value));
        }
        return null;
    }

    private static PrimaryKeyValue convertPrimaryKeyValue(Class<?> type, Object value) {
        if (value == null) {
            return null;
        }
        String typeName = type.getTypeName();
        TsFieldTypeEnum typeEnum = TsFieldTypeEnum.get(typeName);
        if(typeEnum == null) {
            throw new TypeNotSupportException(typeName);
        }
        if(typeEnum.equals(TsFieldTypeEnum.DATE)) {
            Date date = (Date) value;
            return PrimaryKeyValue.fromLong(date.getTime());
        }
        if(typeEnum.equals(TsFieldTypeEnum.BOOLEAN)) {
            return PrimaryKeyValue.fromColumn(ColumnValue.fromBoolean(Boolean.parseBoolean(String.valueOf(value))));
        }
        if(typeEnum.equals(TsFieldTypeEnum.BYTE)) {
            return PrimaryKeyValue.fromColumn(ColumnValue.fromBinary((byte[]) value));
        }
        if(typeEnum.equals(TsFieldTypeEnum.DOUBLE)) {
            return PrimaryKeyValue.fromColumn(ColumnValue.fromDouble(Double.parseDouble(String.valueOf(value))));
        }
        if(typeEnum.equals(TsFieldTypeEnum.LONG)) {
            return PrimaryKeyValue.fromLong(Long.parseLong(String.valueOf(value)));
        }
        if(typeEnum.equals(TsFieldTypeEnum.BIGDECIMAL)) {
            BigDecimal res = (BigDecimal) value;
            return PrimaryKeyValue.fromColumn(ColumnValue.fromDouble(res.doubleValue()));
        }
        if(typeEnum.equals(TsFieldTypeEnum.INTEGER)) {
            return PrimaryKeyValue.fromLong(Long.parseLong(String.valueOf(value)));
        }
        if(typeEnum.equals(TsFieldTypeEnum.STRING)) {
            return PrimaryKeyValue.fromString(String.valueOf(value));
        }
        PrimaryKeyValue.fromColumn(convertColumnValue(type, value));
        return null;
    }

    private  static Object convertColumnToObj(Class<?> type, ColumnValue value){
        if(value == null) {
            return null;
        }
        String typeName = type.getTypeName();
        TsFieldTypeEnum typeEnum = TsFieldTypeEnum.get(typeName);
        if(typeEnum == null) {
            throw new TypeNotSupportException(typeName);
        }
        if(typeEnum.equals(TsFieldTypeEnum.DATE)) {
            return new Date(value.asLong());
        }
        if(typeEnum.equals(TsFieldTypeEnum.BOOLEAN)) {
            return value.asBoolean();
        }
        if(typeEnum.equals(TsFieldTypeEnum.BYTE)) {
            return Byte.valueOf(String.valueOf(value.asLong()));
        }
        if(typeEnum.equals(TsFieldTypeEnum.DOUBLE)) {
            return value.asDouble();
        }
        if(typeEnum.equals(TsFieldTypeEnum.LONG)) {
            return value.asLong();
        }
        if(typeEnum.equals(TsFieldTypeEnum.BIGDECIMAL)) {
            return new BigDecimal(String.valueOf(value.asDouble()));
        }
        if(typeEnum.equals(TsFieldTypeEnum.INTEGER)) {
            return Integer.valueOf(String.valueOf(value.asLong()));
        }
        if(typeEnum.equals(TsFieldTypeEnum.STRING)) {
            return value.asString();
        }
        return null;
    }

    private static boolean isSnake(Class<?> clazz) {
        TsTable tsTable = clazz.getAnnotation(TsTable.class);
        return tsTable.propertySnakeCase();
    }

    private static String translateFieldName(String fileName) {
        PropertyNamingStrategy.SnakeCaseStrategy aa = new PropertyNamingStrategy.SnakeCaseStrategy();
        return aa.translate(fileName);
    }

}
